package com.g.ocache.impl.store;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import com.g.ocache.Cache;
import com.g.ocache.CacheValidator;
import com.g.ocache.Store;
import com.g.ocache.ValidationException;


/**
 * 
 * @author Jingqi Xu
 */
public final class BasicStore<K, V> extends AbstractStore<K, V> {
	//
	protected static final int DEFAULT_INITIAL_CAPACITY = 128;
	
	//
	private final HashMap<K, V> values;
	private final ReentrantReadWriteLock lock;
	
	/**
	 * 
	 */
	public BasicStore(Cache<K, V> cache) {
		this(cache, DEFAULT_INITIAL_CAPACITY, false);
	}
	
	public BasicStore(Cache<K, V> cache, int initialCapacity, boolean fair) {
		//
		super(cache);
		
		//
		this.lock = new ReentrantReadWriteLock(fair);
		this.values = new HashMap<K, V>(initialCapacity);
	}
	
	/**
	 * 
	 */
	protected void doStart() {
		// NOP
	}

	protected void doStop() {
		// NOP
	}
	
	/**
	 * 
	 */
	public final boolean isFair() {
		return lock.isFair();
	}
	
	public void lock(boolean exclusive) {
		final Lock l = exclusive ? this.lock.writeLock() : this.lock.readLock();
		l.lock();
	}
	
	public void unlock(boolean exclusive) {
		final Lock l = exclusive ? this.lock.writeLock() : this.lock.readLock();
		l.unlock();
	}
	
	/**
	 * 
	 */
	public int size() {
		lock(false);
		try {
			return values.size();
		} finally {
			unlock(false);
		}
	}
	
	public boolean containsKey(K key) {
		lock(false);
		try {
			return this.values.containsKey(key);
		} finally {
			unlock(false);
		}
	}
	
	public void clear() {
		lock(true);
		try {
			this.values.clear();
		} finally {
			unlock(true);
		}
	}
	
	public V remove(K key) {
		lock(true);
		try {
			return this.values.remove(key);
		} finally {
			unlock(true);
		}
	}
	
	public V get(K key) {
		lock(false);
		try {
			return this.values.get(key);
		} finally {
			unlock(false);
		}
	}
	
	public Map<K, V> getAll() {
		lock(false);
		try {
			Map<K, V> r = new HashMap<K, V>(this.values.size());
			r.putAll(this.values);
			return r;
		} finally {
			unlock(false);
		}
	}

	public V put(K key, V value) {
		lock(true);
		try {
			//
			final CacheValidator<K, V> validator = getCache().getValidator();
			if(validator != null) {
				V existing = this.values.get(key);
				validator.validate(key, existing, value);
			}
			
			//
			return this.values.put(key, value);
		} finally {
			unlock(true);
		}
	}
	
	public V putIfAbsent(K key, V value) {
		lock(true);
		try {
			//
			V r = this.values.get(key);
			if(r == null) {
				//
				final CacheValidator<K, V> validator = getCache().getValidator();
				if(validator != null) {
					validator.validate(key, null, value);
				}
				
				//
				this.values.put(key, value);
			}
			return r;
		} finally {
			unlock(true);
		}
	}

	public Map<K, V> putAll(Map<K, V> values) {
		Map<K, V> r = new HashMap<K, V>(values.size());
		lock(true);
		try {
			for(K key : values.keySet()) {
				try {
					V existing = put(key, values.get(key)); 
					r.put(key, existing);
				} catch(ValidationException ve) {
					// If validator threw an exception, that exception can be found in "Reject" event
				}
			}
		} finally {
			unlock(true);
		}
		return r;
	}
	
	/**
	 * 
	 */
	public static class FactoryImpl<K, V> implements Factory<K, V> {
		//
		private boolean fair;
		private int initialCapacity = DEFAULT_INITIAL_CAPACITY;

		/**
		 * 
		 */
		public Store<K, V> create(Cache<K, V> cache) {
			return new BasicStore<K, V>(cache, initialCapacity, fair);
		}

		/**
		 * 
		 */
		public boolean isFair() {
			return fair;
		}

		public void setFair(boolean fair) {
			this.fair = fair;
		}

		public int getInitialCapacity() {
			return initialCapacity;
		}

		public void setInitialCapacity(int initialCapacity) {
			this.initialCapacity = initialCapacity;
		}
	}
}
